package com.nepu.gmall.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.nepu.gmall.realtime.app.func.MyBroadcastFunction;
import com.nepu.gmall.realtime.app.func.MyPhoenixSink;
import com.nepu.gmall.realtime.bean.TableProcess;
import com.nepu.gmall.realtime.util.KafkaUtils;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


/**
 * 本类的主要目的是将DIM层的数据写入到Phoenix中
 *  数据的流向：
 *      前端埋点 --> nginx --> 业务服务器 --> mysql --> maxwell --> kafka --> flink --> phoenix
 *      mock --> mysql --> maxwell --> kafka(zookeeper) --> DimSinkApp --> phoenix --> hbase(hdfs)
 * @author chenshuaijun
 * @create 2023-02-16 14:40
 */
public class DimSinkApp {

    public static void main(String[] args) throws Exception {

        // TODO 1、首先是加载环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 生产环境下是一定不会将任务的并行度设置为1的，这里具体的设置是和我们等下要读取的kafka的相应的主题的分区的个数相同
        env.setParallelism(1);
        // 设置checkpoint的信息：设置checkpoint的间隔是5分钟,并且checkpoint的级别是精确一次性
        /*env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的超时时间是10分钟
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        // 设置外部检查点。可以将检查点的元数据信息定期写入外部系统，这样当job失败时，检查点不会被清除。这样如果job失败，可以从检查点恢复job。
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置checkpoint的重启的策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)));
        // 设置两个checkpoint之间的最小的间隔时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // 设置状态后端: 设置状态后端为内存级别
        env.setStateBackend(new HashMapStateBackend());
        // 设置checkpoint的存储的路径
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/checkpoint");
        // 因为我们的HDFS只有atguigu用户才能够操作，所以要将用户设置为atguigu
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/
        // TODO 2、读取数据源
        // 读取对应kafka的topic
        String topic = "topic_db";
        String group_id = "DimSinkApp";
        DataStreamSource<String> dataStreamSource = env.addSource(KafkaUtils.getKafkaConsumer(topic, group_id));
        // TODO 3、清洗并且转换数据结构
        // 这里主要清洗的是那些不符合JSON格式的数据，以及那些是start、complete和delete的数据,使用的算子是flatMap，它可以同时保留filter和map的功能
        SingleOutputStreamOperator<JSONObject> mainStream = dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                // 判断当前数据是否是JSON格式
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    // 如果当前的数据是Json格式，判断对应的类型是不是我们需要的类型
                    String type = jsonObject.getString("type");
                    if (!type.equals("bootstrap-start") && !type.equals("bootstrap-complete") && !type.equals("delete")) {
                        out.collect(jsonObject);
                    }
                } catch (Exception e) {
                    System.out.println("这条数据不是json类型：" + value);
                }
            }
        });
        // TODO 4、使用flinkCDC动态的读取维度的数据：我们需要先创建一个数据库用于存储动态表清单，之后开启这个库的binlog这样就可以使用flinkCDC进行读取了
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall_config")
                .tableList("gmall_config.table_process")
                .username("root")
                .password("12345678")
                .deserializer(new JsonDebeziumDeserializationSchema()) // 解码的时候将对应的数据转换成json格式，方便我们后续的操作
                .startupOptions(StartupOptions.initial()) // 这个表示的是我们每次读取数据是从什么地方读取，我们这里设置成从头读取，其含义就是每次都全量的读取整个表
                .build();
        // TODO 4.1将数据封装成流
        DataStreamSource<String> mysqlSourceStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source");
        // TODO 4.2将流封装成广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("mysql_broadcase", String.class, TableProcess.class);

        BroadcastStream<String> mysqlStreamBroadcast = mysqlSourceStream.broadcast(mapStateDescriptor);
        // TODO 5、双流连接
        BroadcastConnectedStream<JSONObject, String> connectedStream = mainStream.connect(mysqlStreamBroadcast);
        // TODO 6、过滤数据
        SingleOutputStreamOperator<JSONObject> dimDs = connectedStream.process(new MyBroadcastFunction(mapStateDescriptor));
        // TODO 7、将数据写入到Phoenix
        dimDs.print("写入到Phoenix中的数据：");

        dimDs.addSink(new MyPhoenixSink());
        // TODO 8、执行
        env.execute();

    }
}
